package bqds

import (
	// "ctpdefine"
	"context"
	"encoding/json"
	"fmt"
	"log"
	"os"
	"strconv"
	"strings"
	"time"

	ctpdefine "gitee.com/boxigg/ctpdefine"

	// "github.com/pkg/errors"
	"github.com/sirupsen/logrus"

	// "github.com/go-redis/redis/v8"
	"gitee.com/boxigg/ctpredis"

	"database/sql"

	_ "github.com/lib/pq"

	"gorm.io/driver/postgres"
	"gorm.io/gorm"
	"gorm.io/gorm/logger"
	"gorm.io/gorm/schema"
)

// var (
// 	rhost     = "114.132.166.195"
// 	rport     = 6379
// 	user      = "lzzzkz"
// 	rpassword = "steward1!"
// 	rdb       = ctpredis.COMMONDB
// )

type PgDb struct {
	host   string
	port   int
	user   string
	dbname string

	Db          *sql.DB
	cr          *ctpredis.CtpRedis
	tickStmt    *sql.Stmt
	chClose     chan int
	preMargin   map[string]float64
	prePosition map[string]float64
	preOrder    map[string]string
	preTrade    map[string]string
	redisCtxt   context.Context
	GORMpg      *gorm.DB
	savedBar    bool
	MainIID     []string
	redisDB     int
}

func NewPgDb(host, usr, pwd, dbname string, port int) *PgDb {
	pgsqlInfo := fmt.Sprintf("host=%s port=%d user=%s "+
		"password=%s dbname=%s sslmode=disable",
		host, port, usr, pwd, dbname)

	db, err := sql.Open("postgres", pgsqlInfo)
	if err != nil {
		logrus.Infof("连接： %s ：错误::\n", pgsqlInfo)
		panic(err)
	}

	err = db.Ping()
	if err != nil {
		panic(err)
	}
	logrus.Info("Successfully connected to postresql: !", pgsqlInfo)
	// InitPgDB("../../ds/sql.yml", db)

	pd := new(PgDb)
	pd.redisCtxt = context.Background()
	pd.savedBar = false
	pd.Db = db
	pd.host = host
	pd.port = port
	pd.user = usr
	pd.dbname = dbname

	// 初始化 tickStmt
	pd.tickStmt, err = pd.Db.Prepare(`insert into tick.tmpl( "DateTime", "ExchangeID",
	    "HighestPrice", "LowestPrice","Volume", "OpenInterest", "TradingDay", 
		"InstrumentID", "LastPrice",	"PreSettlementPrice", "PreClosePrice", 
		"PreOpenInterest",	"OpenPrice", "Turnover", "ClosePrice", "SettlementPrice",
		"UpperLimitPrice", "LowerLimitPrice", 	"UpdateMillisec", "UpdateTime",
		"BidPrice1", "BidVolume1", "AskPrice1",	"AskVolume1","AveragePrice")
		values($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23,$24,$25)`)
	if err != nil {
		panic(err)
	}
	pd.preMargin = make(map[string]float64)
	pd.prePosition = make(map[string]float64)
	pd.preOrder = make(map[string]string)
	pd.preTrade = make(map[string]string)

	newLogger := logger.New(
		log.New(os.Stdout, "\r\n", log.LstdFlags), // io writer
		logger.Config{
			SlowThreshold:             5000 * time.Millisecond, // 慢 SQL 阈值, 200ms
			LogLevel:                  logger.Warn,             // 日志级别
			IgnoreRecordNotFoundError: true,                    // 忽略ErrRecordNotFound（记录未找到）错误
			Colorful:                  true,                    // 彩色打印
		},
	)
	var schemaName = "bar" // 模式
	// var err error
	pgMin := "postgresql://postgres:steward1!@192.168.10.5:5432"
	pgMin = pgsqlInfo
	pd.GORMpg, err = gorm.Open(postgres.Open(pgMin), &gorm.Config{
		SkipDefaultTransaction: true, // 跳过事务一致性检查,默认false,设为 true可提升性能,但不保证数据准确性
		NamingStrategy: schema.NamingStrategy{
			TablePrefix:   schemaName + ".", //+ ".t_", // 表名前置 User => t_users
			SingularTable: true,             // 是不表名复数, false:T_users  true:T_user
		},
		NowFunc:                                  func() time.Time { return time.Now() }, // 替换默认 CreateAt UpdateAt 的时间创建函数
		DisableForeignKeyConstraintWhenMigrating: true,                                   // 禁止创建物理外键, 推荐设为true, 因物理外键<非常>影响性能, 多用逻辑外键(代码中体现外键关系)
		Logger:                                   newLogger,
	})
	if err != nil {
		logrus.Fatal("pgMin 配置错误:", err)
	}
	// pd.GORMpg.Exec("CREATE SCHEMA IF NOT EXISTS " + schemaName)
	// pd.chClose=make(chan int)
	return pd
}

func (pd *PgDb) NewCtpRedis(ip, auth string, port, db int) (err error) {
	pd.cr, err = ctpredis.NewCtpRedis(ip, auth, port, db)
	pd.redisDB = db
	return
}

// 关闭postgresql server
func (pd *PgDb) Close() {
	pd.Db.Close()
	pd.chClose <- 1
}

func (pd *PgDb) GetMainIID() []string {
	pl := pd.cr.Rdb
	// pl := rdb.TxPipeline()
	pl.Do(pd.cr.Ctx, "select", 2)
	// keys,err:=pl.HGetAll(pd.cr.Ctx,"main_future_dict").Result()
	// if err != nil {
	// 	return nil
	// }
	miid, err := pl.HVals(pd.cr.Ctx, "main_future_dict").Result()
	if err != nil {
		return nil
	}
	pl.Do(pd.cr.Ctx, "select", pd.redisDB)
	// _, err = pl.Exec(pd.cr.Ctx)
	// if err != nil {
	// 	return nil
	// }
	logrus.Info("miid:=", miid[1], len(miid))
	for _, v := range miid {
		s := strings.Split(v, ",")
		pd.MainIID = append(pd.MainIID, s...)
	}
	return pd.MainIID

}

// 收到redis publish 的 行情 tick
func (pd *PgDb) OnTick() {
	ticks := 0
	var chnls []string
	for _, v := range pd.MainIID {
		chnls = append(chnls, ctpredis.TICK+v)
		// logrus.Info(ctpredis.TICK + v)
	}
	logrus.Info("subscribed ticks:", chnls)
	sub := pd.cr.Rdb.PSubscribe(pd.redisCtxt, chnls...) // ctpredis.TICK + "*501") // ctpredis.TICK  "tick."
	go func() {
		defer sub.Close()
		cm := sub.Channel()
		for {
			select {
			case <-pd.chClose:
				{
					logrus.Info("return ")
					return
				}
			case msg := <-cm:
				if msg == nil {
					logrus.Info("no msg")
					break
				}
				//保存数据到 postgresql
				// tick, _ := json.Marshal(msg.Payload)
				// logrus.Info("msg.Payload:\n", msg.Payload)
				// logrus.Info(tick)
				// 不可用 go 执行 保存，否则会造成
				if msg.Payload == "" {
					break
				}
				ticks += 1
				// if ticks>=10000{
				// 	logrus.Info("save ticks:",ticks)
				// }
				pd.saveTick([]byte(msg.Payload))
				//
			}
		}

	}()

}
func (pd *PgDb) saveTick(bs []byte) {
	// logrus.Info("saveTick: ", bs == nil, bs)
	var tick map[string]interface{}
	err := json.Unmarshal([]byte(bs[:]), &tick)
	if err != nil {
		logrus.Info("", err)
	}

	_, err = pd.tickStmt.Exec(
		tick["TradingDay"].(string)+" "+tick["UpdateTime"].(string)+"."+strconv.FormatFloat(tick["UpdateMillisec"].(float64), 'f', 0, 32), tick["ExchangeID"],
		tick["HighestPrice"], tick["LowestPrice"], tick["Volume"], tick["OpenInterest"], tick["TradingDay"],
		tick["InstrumentID"], tick["LastPrice"], tick["PreSettlementPrice"],
		tick["PreClosePrice"], tick["PreOpenInterest"], tick["OpenPrice"], tick["Turnover"], 0,
		0, tick["UpperLimitPrice"], tick["LowerLimitPrice"], tick["UpdateMillisec"], tick["UpdateTime"],
		tick["BidPrice1"], tick["BidVolume1"], tick["AskPrice1"], tick["AskVolume1"], tick["AveragePrice"])
	if err != nil {
		logrus.Info(tick["TradingDay"].(string)+" "+tick["UpdateTime"].(string)+"."+strconv.FormatFloat(tick["UpdateMillisec"].(float64), 'f', 0, 32), tick["ExchangeID"],
			tick["HighestPrice"], tick["LowestPrice"], tick["Volume"], tick["OpenInterest"], tick["TradingDay"],
			tick["InstrumentID"], tick["ExchangeID"], tick["LastPrice"], tick["PreSettlementPrice"],
			tick["PreClosePrice"], tick["PreOpenInterest"], tick["OpenPrice"], tick["Turnover"], 0,
			0, tick["UpperLimitPrice"], tick["LowerLimitPrice"], tick["UpdateMillisec"], tick["UpdateTime"],
			tick["BidPrice1"], tick["BidVolume1"], tick["AskPrice1"], tick["AskVolume1"], tick["AveragePrice"])
		logrus.Info(err)
	}

}

func (pd *PgDb) OnAcc() {
	sub := pd.cr.Subscribe(ctpredis.ACCOUNT+"*", ctpredis.POSITION+"*",
		ctpredis.ORDER+"*", ctpredis.TRADE+"*", ctpredis.EVENT_ACCOUNT+"*",
		ctpredis.EVENT_ORDER+"*", ctpredis.EVENT_POSITION+"*", ctpredis.EVENT_ORDER+"*",
		ctpredis.EVENT_POSITION_DETAIL+"*", ctpredis.EVENT_TRADE+"*")
	go func() {
		defer sub.Close()
		cm := sub.Channel()
		for {
			select {
			case <-pd.chClose:
				{
					logrus.Info("return ")
					return
				}
			case msg := <-cm:
				if msg == nil {
					logrus.Info("no msg")
					break
				}
				pl := msg.Payload
				// ptn := msg.Pattern
				//保存数据到 postgresql
				var data map[string]interface{}
				if strings.Contains(pl, "\x00") {
					logrus.Info("origin payload:\n", pl)
					pl = strings.ReplaceAll(pl, "\x00", "")
					logrus.Info("new payload:\n", pl)
				}
				json.Unmarshal([]byte(pl), &data)
				// jsm, _ := json.Marshal(pl)
				channel := msg.Channel
				t, _ := strings.Split(channel, ".")[0], strings.Split(channel, ".")[1]
				logrus.Info("msg channel:", channel, t)
				switch t + "." {
				case ctpredis.ACCOUNT:
					// logrus.Info(pl)
					pd.saveAsset(data)
				case ctpredis.TRADE:
					logrus.Info("ctpredis.TRADE:\n", data)
					pd.saveTrade(data)
				case ctpredis.ORDER:
					logrus.Info("ctpredis.ORDER:\n", data)
					pd.saveOrder(data)
				case ctpredis.POSITION:
					pd.savePosition(data)
				case ctpredis.EVENT_ACCOUNT:
					// json.Unmarshal([]byte(pl[2:]), &data)
					// logrus.Info(data)
					// logrus.Info(pl)
					pd.saveAsset(data)
				case ctpredis.EVENT_TRADE:
					logrus.Info("ctpredis.EVENT_TRADE:\n", data)
					pd.saveTrade(data)
				case ctpredis.EVENT_POSITION:
					logrus.Info("ctpredis.EVENT_POSITION:\n", data)
					pd.savePosition(data)
				case ctpredis.EVENT_ORDER:
					logrus.Info("ctpredis.EVENT_ORDER:\n", data)
					pd.saveOrder(data)
				case ctpredis.EVENT_POSITION_DETAIL:
					logrus.Info("ctpredis.EVENT_POSITION_DETAIL:\n", data)
					pd.savePositionDetail(data)
				case ctpredis.EVENT_CTPORDER:
					logrus.Info("ctpredis.EVENT_CTPORDER:\n", data)
					pd.saveCTPOrder(data)
				}

				// logrus.Info("msg.Payload:\n", msg.Channel, ptn)
				// logrus.Info(data)
				// go pd.saveTick([]byte(msg.Payload))
				//
			}
		}

	}()
}

func (pd *PgDb) saveAsset(df map[string]interface{}) {
	// var ast map[string]interface{}
	// json.Unmarshal(bs, &ast)
	// fmt.Println("pd.saveAsset:df\n", df)
	if df["BrokerID"] == nil {
		return
	}
	id := df["BrokerID"].(string) + df["AccountID"].(string)
	b := pd.preMargin[id] // 只有当保证金变动后再保存
	if b != df["FrozenCommission"].(float64) || true {
		pd.preMargin[id] = df["FrozenCommission"].(float64)

		sql := ` insert into account.acc_finance("AccountID","BrokerID","InvestorID","Credit","Deposit","Withdraw",
		"PreDeposit","PreMargin","PreBalance","CurrMargin","CashIn","Balance","Available","Commission","CloseProfit",
		"PositionProfit","FrozenCash","FrozenCommission","ExchangeMargin","DeliveryMargin","ExchangeDeliveryMargin",
		"ReserveBalance","TradingDay")values($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23)
	`
		stmt, err := pd.Db.Prepare(sql)

		if err != nil {
			logrus.Info(err)
			logrus.Info(sql, "saveAsset")
		}
		defer stmt.Close()
		stmt.Exec(df["AccountID"], df["BrokerID"], df["AccountID"],
			df["Credit"], df["Deposit"], df["Withdraw"], df["PreDeposit"], df["PreMargin"], df["PreBalance"],
			df["CurrMargin"], df["CashIn"], df["Balance"], df["Available"], df["Commission"], df["CloseProfit"],
			df["PositionProfit"], df["FrozenCash"], df["FrozenCommission"], df["ExchangeMargin"],
			df["DeliveryMargin"], df["ExchangeDeliveryMargin"], df["ReserveBalance"], df["TradingDay"])
		// pd.Db.Exec(sql)
	}
}

func (pd *PgDb) savePosition(df map[string]interface{}) {
	// var ast map[string]interface{}
	// json.Unmarshal(bs, &ast)
	// fmt.Println("pd.savePosition:df\n", df)
	// logrus.Info(df["BrokerID"], df["InvestorID"], df["InstrumentID"],df["PositionDirection"])
	id := df["BrokerID"].(string) + df["InvestorID"].(string) + df["InstrumentID"].(string) + strconv.FormatFloat(df["PositionDirection"].(float64), 'f', -1, 64)

	b := pd.prePosition[id]
	if b != df["Position"].(float64) {
		pd.prePosition[id] = df["Position"].(float64)
		sql := `insert into account.position("BrokerID","InvestorID","InstrumentID","CloseAmount","TradingDay",
		"CloseProfit","CloseProfitByDate","CloseProfitByTrade","TodayPosition","YdPosition","Position",
		"UseMargin","PreMargin","PositionCost","FrozenMargin","FrozenCash","PositionCostOffset",
		"FrozenCommission","CashIn","Commission","PreSettlementPrice","PosiDirection","OpenCost",
		"OpenVolume","CloseVolume","ExchangeMargin","OpenAmount")
		values($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23,$24,$25,$26,$27)`
		stmt, err := pd.Db.Prepare(sql)
		if err != nil {
			logrus.Info(df, "::savePosition")
			logrus.Info("savePosition error", err)
		}
		defer stmt.Close()

		if df["TradingDay"] == nil {
			df["TradingDay"] = df["PositionDate"]
		}
		_, err = stmt.Exec(df["BrokerID"], df["InvestorID"], df["InstrumentID"], df["CloseAmount"], df["TradingDay"], df["CloseProfit"],
			df["CloseProfitByDate"], df["CloseProfitByTrade"], df["TodayPosition"], df["YdPosition"], df["Position"], df["UseMargin"], df["PreMargin"],
			df["PositionCost"], df["FrozenMargin"], df["FrozenCash"], df["PositionCostOffset"], df["FrozenCommission"], df["CashIn"], df["Commission"],
			df["PreSettlementPrice"], df["PositionDirection"], df["OpenCost"], df["OpenVolume"], df["CloseVolume"], df["ExchangeMargin"], df["OpenAmount"])
		if err != nil {
			logrus.Info(df, "::savePosition")
			logrus.Error("ds.savePosition error:", err)
		}
	}
}

func (pd *PgDb) savePositionDetail(df map[string]interface{}) {
	// var ast map[string]interface{}
	// json.Unmarshal(bs, &ast)
	// fmt.Println("pd.savePosition:df\n", df)
	// logrus.Info(df["BrokerID"], df["InvestorID"], df["InstrumentID"],df["PositionDirection"])
	id := df["BrokerID"].(string) + df["InvestorID"].(string) +
		df["InstrumentID"].(string) + strconv.FormatFloat(df["PositionDirection"].(float64), 'f', -1, 64)

	b := pd.prePosition[id]
	if b != df["Position"].(float64) {
		pd.prePosition[id] = df["Position"].(float64)
		sql := `insert into account.position("BrokerID","InvestorID","InstrumentID","CloseAmount","TradingDay",
		"CloseProfit","CloseProfitByDate","CloseProfitByTrade","TodayPosition","YdPosition","Position",
		"UseMargin","PreMargin","PositionCost","FrozenMargin","FrozenCash","PositionCostOffset",
		"FrozenCommission","CashIn","Commission","PreSettlementPrice","PosiDirection","OpenCost",
		"OpenVolume","CloseVolume","ExchangeMargin","OpenAmount")
		values($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23,$24,$25,$26,$27)`
		stmt, err := pd.Db.Prepare(sql)
		if err != nil {
			logrus.Info(df, "::savePosition")
			logrus.Info("savePosition error", err)
		}
		defer stmt.Close()

		if df["TradingDay"] == nil {
			df["TradingDay"] = df["PositionDate"]
		}
		_, err = stmt.Exec(df["BrokerID"], df["InvestorID"], df["InstrumentID"], df["CloseAmount"], df["TradingDay"], df["CloseProfit"],
			df["CloseProfitByDate"], df["CloseProfitByTrade"], df["TodayPosition"], df["YdPosition"], df["Position"], df["UseMargin"], df["PreMargin"],
			df["PositionCost"], df["FrozenMargin"], df["FrozenCash"], df["PositionCostOffset"], df["FrozenCommission"], df["CashIn"], df["Commission"],
			df["PreSettlementPrice"], df["PositionDirection"], df["OpenCost"], df["OpenVolume"], df["CloseVolume"], df["ExchangeMargin"], df["OpenAmount"])
		if err != nil {
			logrus.Info(df, "::savePosition")
			logrus.Error("ds.savePosition error:", err)
		}
	}
}

func (pd *PgDb) saveTrade(df map[string]interface{}) {
	// var ast map[string]interface{}
	// json.Unmarshal(bs, &ast)
	// fmt.Println("pd.saveTrade:df\n", df)
	bs, err := json.Marshal(df)
	if err != nil {
		logrus.Error(err)
	}
	td := ctpdefine.TradingAccountField{}
	json.Unmarshal(bs[:], &td)
	logrus.Info("td:", td)

	logrus.Info(df)
	id := df["BrokerID"].(string) + df["InvestorID"].(string) + df["ExchangeID"].(string) + df["TradeID"].(string)
	b := pd.preTrade[id]
	if b != df["TradeTime"].(string) {
		pd.preTrade[id] = df["TradeTime"].(string)
		sql := `insert into account."trade"("InvestorID","BrokerID","InstrumentID","ExchangeID","TradeID","OrderRef",
		"OrderSysID","TradeDate","TradeTime","TradingDay","Direction","OffsetFlag","Price","Volume")
		values($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14)`
		stmt, err := pd.Db.Prepare(sql)

		if err != nil {
			logrus.Info(sql, "::saveTrade:error:", err)
		}
		defer stmt.Close()
		_, err = stmt.Exec(df["InvestorID"], df["BrokerID"], df["InstrumentID"], df["ExchangeID"], df["TradeID"], df["OrderRef"],
			df["OrderSysID"], df["TradeDate"], df["TradeTime"], df["TradingDay"], df["Direction"], df["OffsetFlag"], df["Price"], df["Volume"])
		if err != nil {
			logrus.Error("ds.saveTrade error of stmt.Exec:", err)
			logrus.Info(df)
		}
	}

	// fmt.Println(df)
}
func (pd *PgDb) saveOrder(df map[string]interface{}) {
	// logrus.Info(df["InvestorID"].(string))
	// var df map[string]interface{}
	fmt.Println("pd.saveOrder:order::\n", df)

	id := df["BrokerID"].(string) + df["InvestorID"].(string) +
		df["ExchangeID"].(string) + strconv.FormatFloat(df["SessionID"].(float64), 'f', -1, 64)
	b := pd.preOrder[id]
	if b != df["InsertTime"].(string) {
		pd.preOrder[id] = df["InsertTime"].(string)
		sql := `insert into account."Order"("InvestorID","BrokerID","InstrumentID",
		"ExchangeID","OrderStatus","OrderRef","OrderSysID","InsertDate",
		"InsertTime","CancelTime","Direction","CombOffsetFlag","LimitPrice","RequestID",
		"VolumeTotalOriginal","VolumeTraded","VolumeLeft","FrontID","SessionID")
		values($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19)` //"LastTradeTime",  ,$20
		stmt, err := pd.Db.Prepare(sql)
		if err != nil {
			logrus.Info(err)
			logrus.Info(sql, "::saveOrder prepare error::", err)
			return
		}
		defer stmt.Close()
		_, err = stmt.Exec(df["InvestorID"], df["BrokerID"], df["InstrumentID"], df["ExchangeID"], df["OrderStatus"],
			df["OrderRef"], df["OrderSysID"], df["InsertDate"], df["InsertTime"], df["CancelTime"],
			df["Direction"], df["OffsetFlag"], df["LimitPrice"], df["RequestID"], df["VolumeTotalOriginal"],
			df["VolumeTraded"], df["VolumeLeft"], df["FrontID"], df["SessionID"]) // df["LastTradeTime"],
		if err != nil {
			logrus.Error("ds.saveOrder:", err)
			logrus.Info(df["InvestorID"], df["BrokerID"], df["InstrumentID"], df["ExchangeID"], df["OrderStatus"],
				df["OrderRef"], df["OrderSysID"], df["InsertDate"], df["InsertTime"], df["CancelTime"],
				df["Direction"], df["OffsetFlag"], df["LimitPrice"], df["RequestID"], df["VolumeTotalOriginal"],
				df["VolumeTraded"], df["VolumeLeft"], df["FrontID"], df["SessionID"]) // df["LastTradeTime"],
		}

	}

}

func (pd *PgDb) saveCTPOrder(df map[string]interface{}) {
	// logrus.Info(df["InvestorID"].(string))
	// var df map[string]interface{}

	vl := int(df["VolumeTotalOriginal"].(float64)) - int(df["VolumeTraded"].(float64))
	fmt.Println("pd.saveCTPOrder:order::\n", df)

	id := df["BrokerID"].(string) + df["InvestorID"].(string) +
		df["ExchangeID"].(string) + strconv.FormatFloat(df["SessionID"].(float64), 'f', -1, 64)
	b := pd.preOrder[id]
	if b != df["InsertTime"].(string) || true {
		pd.preOrder[id] = df["InsertTime"].(string)
		sql := `insert into account."Order"("InvestorID","BrokerID","InstrumentID",
		"ExchangeID","OrderStatus","OrderRef","OrderSysID","InsertDate",
		"InsertTime","CancelTime","Direction","CombOffsetFlag","LimitPrice","RequestID",
		"VolumeTotalOriginal","VolumeTraded","VolumeLeft","FrontID","SessionID")
		values($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19)` //"LastTradeTime",  ,$20
		stmt, err := pd.Db.Prepare(sql)
		if err != nil {
			logrus.Info(err)
			logrus.Info(sql, "::saveOrder prepare error::", err)
			return
		}

		defer stmt.Close()
		_, err = stmt.Exec(df["InvestorID"], df["BrokerID"], df["InstrumentID"],
			df["ExchangeID"], df["OrderStatus"], df["OrderRef"], df["OrderSysID"], df["InsertDate"],
			df["InsertTime"], df["CancelTime"], df["Direction"], df["CombOffsetFlag"], df["LimitPrice"], df["RequestID"],
			df["VolumeTotalOriginal"], df["VolumeTraded"], vl, df["FrontID"], df["SessionID"]) // df["LastTradeTime"],
		if err != nil {
			logrus.Error("ds.saveOrder:", err)
			logrus.Info(df["InvestorID"], df["BrokerID"], df["InstrumentID"], df["ExchangeID"], df["OrderStatus"],
				df["OrderRef"], df["OrderSysID"], df["InsertDate"], df["InsertTime"], df["CancelTime"],
				df["Direction"], df["OffsetFlag"], df["LimitPrice"], df["RequestID"], df["VolumeTotalOriginal"],
				df["VolumeTraded"], vl, df["FrontID"], df["SessionID"]) // df["LastTradeTime"],
		}

	}

}

func (pd *PgDb) LoadTick(inst string, sd, ed string) []ctpdefine.StraTickField {
	//inst: instrument id , sd: start datetime, ed: end datetime
	ticks := make([]ctpdefine.StraTickField, 100)
	sql := `select  "DateTime", "PreSettlementPrice", "PreClosePrice", "PreOpenInterest", 
	"UpperLimitPrice", "LowerLimitPrice","OpenPrice", 	"HighestPrice", "LowestPrice", 
	 "Volume", "Turnover", "OpenInterest","AveragePrice" ,  "LastPrice",
	"BidPrice1", "BidVolume1", "AskPrice1", "AskVolume1"
	from tick.tmpl where "Instrument"='` + inst +
		"' and \"DateTime\" between '" + sd + "' and '" + ed + "'" +
		" order by " + `"DateTime" desc;`
	rows, e := pd.Db.Query(sql)
	if e != nil {
		logrus.Error(e)
		logrus.Info("sql:", sql)
		return nil
	}
	// var row ctpdefine.StraTickField // map[string]interface{}
	// row = ctpdefine.StraTickField{}
	defer rows.Close()
	logrus.Info(rows.Columns())
	for rows.Next() {
		tick := ctpdefine.StraTickField{}
		rows.Scan(&tick)
		logrus.Info(tick)
		ticks = append(ticks, tick)
	}

	return ticks
}

func (pd *PgDb) OnReqLoadTick() {
	var (
		inst   string
		sd, ed string
		prms   map[string]interface{}
	)
	sub := pd.cr.Rdb.PSubscribe(pd.redisCtxt, ctpredis.RQHISTORYTICK+"*")
	go func() {
		defer sub.Close()
		cm := sub.Channel()
		for {
			select {
			case <-pd.chClose:
				{
					logrus.Info("return ")
					return
				}
			case msg := <-cm:
				if msg == nil {
					logrus.Info("no msg")
					break
				}

				if msg.Payload == "" {
					break
				}

				json.Unmarshal([]byte(msg.Payload), &prms)
				inst, sd, ed = prms["inst"].(string), prms["inst"].(string), prms["inst"].(string)
				data := pd.LoadTick(inst, sd, ed)
				pd.cr.Rdb.Publish(pd.redisCtxt, ctpredis.RSPHISTORYTICK, data)

			}
		}

	}()

}

func (pd *PgDb) OnReqDataByFunc(reqInfo map[string]interface{}, reqFunc func(interface{})) {

}
